/*
* Copyright (c) 2009, Active Web Solutions Limited
* All rights reserved.
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are met:
*
* 1- Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
*
* 2- Redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in the
* documentation and/or other materials provided with the distribution.
*
* 3- The name of the author may not be used to endorse or promote products
* derived from this software without specific prior written permission.
*
*
* THIS SOFTWARE IS PROVIDED BY Active Web Solutions Limited ''AS IS'' AND ANY
* EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
* WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
* DISCLAIMED. IN NO EVENT SHALL Active Web Solutions Limited BE LIABLE FOR ANY
* DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
* (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
* LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
* ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
* SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*
*/
package net.aws.jazure;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import javax.xml.parsers.DocumentBuilderFactory;
import javax.xml.parsers.ParserConfigurationException;
import javax.xml.xpath.XPath;
import javax.xml.xpath.XPathConstants;
import javax.xml.xpath.XPathExpressionException;
import javax.xml.xpath.XPathFactory;

import net.aws.jazure.exception.APIRequestFailedException;
import net.aws.jazure.httpconstants.CompConstants;
import net.aws.jazure.httpconstants.HeaderNames;
import net.aws.jazure.httpconstants.QueryParams;
import net.aws.jazure.httpconstants.RequestParams;
import net.aws.jazure.httpconstants.XMLElements;
import net.aws.jazure.util.RestHelper;

import org.apache.log4j.Logger;
import org.restlet.Client;
import org.restlet.data.ChallengeResponse;
import org.restlet.data.ChallengeScheme;
import org.restlet.data.Form;
import org.restlet.data.MediaType;
import org.restlet.data.Method;
import org.restlet.data.Protocol;
import org.restlet.data.Request;
import org.restlet.data.Response;
import org.restlet.data.Status;
import org.restlet.engine.util.Base64;
import org.restlet.representation.DomRepresentation;
import org.restlet.representation.Representation;
import org.restlet.representation.StringRepresentation;
import org.w3c.dom.Document;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;
import org.xml.sax.InputSource;
import org.xml.sax.SAXException;

/**
 * <p>The Queue class.</p>
 * <p>Used to access messages in the queue.</p>
 * 
 * @author dyaa.albakour
 *
 */
public class Queue {

	private String accountName;
	private String queueName;
	private String accountSharedKey;
	private String endPoint;
	
	/**
	 * A constructor with default access modifier.
	 * for package use only.
	 * 
	 * @param endPoint
	 * @param queueName
	 * @param accountName
	 * @param accountSharedKey
	 */
	Queue(String endPoint, String queueName, String accountName, String accountSharedKey) {
		this.endPoint = endPoint;
		this.accountName = accountName;
		this.queueName = queueName;
		this.accountSharedKey = accountSharedKey;
	}
	
	
	/**
	 * create a new queue. 
	 * @return true if the creation was successful or the queue already exists.
	 * @throws APIRequestFailedException If the API request return an unexpected HTTP status code.
	 */
	public boolean create() throws APIRequestFailedException {
		
		// TODO specify the metadata when creating the queue.
		
		Client client = new Client(Protocol.HTTP); 
		ChallengeResponse challengeResponse =  new ChallengeResponse(
	    	ChallengeScheme.HTTP_MS_SHAREDKEY,
	    	accountName,
	    	accountSharedKey);
		
		
		String url = endPoint + "/" + queueName;
		Request request = new Request(Method.PUT, url);		
		
		request.setChallengeResponse(challengeResponse);		
		
		// RESTlet seems not to be setting the content length of the request
		// 	with the PUT method and no content.
		// Azure Queue service is complaining about content length being not specified
		// To overcome this I add to the request a dummy string..	
		request.setEntity(new StringRepresentation("dummy"));	
		
		Response response = client.handle(request);			

		
		// The creation succeeds if the HTTP response was 201 ( success created).	
		Status responseStatus = response.getStatus();
		if(responseStatus.equals(Status.SUCCESS_CREATED) ||
				responseStatus.equals(Status.CLIENT_ERROR_CONFLICT))
			return true;
		else	
			throw new APIRequestFailedException(responseStatus);
	}
	/**
	 * <p>Deletes the current queue.</p>
	 * @return true if the queue is successfully deleted.
	 * @throws APIRequestFailedException If the API request return an unexpected HTTP status code.
	 * 
	 */
	public boolean delete() throws APIRequestFailedException {		
		boolean result = false;
		
		Client client = new Client(Protocol.HTTP); 
		ChallengeResponse challengeResponse =  new ChallengeResponse(
	    	ChallengeScheme.HTTP_MS_SHAREDKEY,
	    	accountName,
	    	accountSharedKey);
		
		String url = endPoint + "/" + queueName;
		Request request = new Request(Method.DELETE, url);
		request.setChallengeResponse(challengeResponse);
		Response response = client.handle(request);
		
		// The deletion succeeds if the HTTP response was 204 (No Content).
		Status responseStatus = response.getStatus();
		if(responseStatus.equals(Status.SUCCESS_NO_CONTENT))
			return true;
		else if(responseStatus.equals(Status.CLIENT_ERROR_NOT_FOUND)
				|| responseStatus.equals(Status.CLIENT_ERROR_CONFLICT) 
				|| responseStatus.equals(Status.CLIENT_ERROR_PRECONDITION_FAILED) )
			return false;
		else
			throw new APIRequestFailedException();
	}
	
	/**
	 * Get the queue properties of this queue ( Metadata + approximate number of messages ).
	 *  
	 * @return QueueProperties The queue properties. 
	 * @throws APIRequestFailedException If the API request return an unexpected HTTP status code.
	 */
	public QueueProperties getProperties() throws APIRequestFailedException{
		
		QueueProperties queueProperties = null;	
		try{					
			// build a REST request to the Azure queue service
			// e.g
			//  http://127.0.0.1:10001/devstoreaccount1/myqueue?comp=metadata
			Client client = new Client(Protocol.HTTP); 
			ChallengeResponse challengeResponse =  new ChallengeResponse(
		    	ChallengeScheme.HTTP_MS_SHAREDKEY,
		    	accountName,
		    	accountSharedKey);			
			
			Map<String, String> parameters= new HashMap<String, String>();
			parameters.put(QueryParams.QueryParamComp, CompConstants.Metadata);
			
			String url = RestHelper.createRequestURI(endPoint + "/" + queueName,parameters);
			
			Request request = new Request(Method.GET, url);
			request.setChallengeResponse(challengeResponse);
			
			// retrieve the headers from the response.
			Response response = client.handle(request);			
			
			if(response.getStatus().equals(Status.SUCCESS_OK)){
				Form headers =(Form)response.getAttributes().get("org.restlet.http.headers");
				
				queueProperties = new QueueProperties();
				
				for(String key: headers.getNames()){				
					int prefixLength = HeaderNames.PrefixForMetadata.length();				
					// check if the parameter is the approximate message count.
					if(key.equalsIgnoreCase(HeaderNames.ApproximateMessagesCount)){
						String value = headers.getFirstValue(key);
						int approximateMessageCount = Integer.parseInt(value);
						queueProperties.setApproximateMessageCount(approximateMessageCount);
					}				
					// check if the parameter is a meta-data value				
					else if ( key.toLowerCase().startsWith(HeaderNames.PrefixForMetadata.toLowerCase()) )
					{
						// work out the meta-data key and add the pair to the meta-data map.
						queueProperties.getMetaData().put(
								key.substring(prefixLength), headers.getFirstValue(key));
					}			
				}			
				return queueProperties;
			}
			else 
				throw new APIRequestFailedException(response.getStatus());
			
		}catch (UnsupportedEncodingException e){
			// Create Request failed
			return null;
		}
		
	}
	
	/**
	 * Set the queue properties of this queue ( Metadata + approximate number of messages ). 
	 * @param queueProperties The queue properties.
	 * @return true if the properties are successfully set.
	 * @throws APIRequestFailedException If the API request return an unexpected HTTP status code.
	 */
	public boolean setProperties(QueueProperties queueProperties) 
		throws APIRequestFailedException{
		try{
			// build a REST request to the Azure queue service
			// e.g
			//  http://127.0.0.1:10001/devstoreaccount1/myqueue?comp=metadata
			Client client = new Client(Protocol.HTTP); 
			ChallengeResponse challengeResponse =  new ChallengeResponse(
		    	ChallengeScheme.HTTP_MS_SHAREDKEY,
		    	accountName,
		    	accountSharedKey);
			
			Map<String, String> parameters= new HashMap<String, String>();
			parameters.put(QueryParams.QueryParamComp, CompConstants.Metadata);
			
			String url = RestHelper.createRequestURI(endPoint + "/" + queueName,parameters);
			
			Request request = new Request(Method.PUT, url);
			request.setChallengeResponse(challengeResponse);			
			
			// add the meta-data to be set in the headers.
			// Note that the prefix should be added to the name of the header.
			Form headers = new Form();
			for(String key: queueProperties.getMetaData().keySet()){
				headers.add(HeaderNames.PrefixForMetadata + key,queueProperties.getMetaData().get(key) );
			}			
			
			request.getAttributes().put("org.restlet.http.headers",headers);			
			request.setEntity(new StringRepresentation("dummy"));	
						
			//now send the request and check for the result
			Response response = client.handle(request);
			
			Status responseStatus = response.getStatus();
		
			if(responseStatus.equals(Status.SUCCESS_NO_CONTENT))
				return true;
			else
				throw new APIRequestFailedException(responseStatus);
		}
		catch(UnsupportedEncodingException ex){
			return false;
		}
		
	}
	
	/**
	 * Puts a message in the queue.
	 * 
	 * @param message The message to store in the queue
	 * @return true if the message is successfully enqueued.
	 * @throws APIRequestFailedException If the API request return an unexpected HTTP status code.
	 */
	public boolean putMessage(QueueMessage message) 
	throws APIRequestFailedException {
		try{
			Client client = new Client(Protocol.HTTP); 
			ChallengeResponse challengeResponse =  new ChallengeResponse( ChallengeScheme.HTTP_MS_SHAREDKEY,
										      accountName,
										      accountSharedKey);
			
			String url = endPoint + "/" + queueName + "/messages";
			Request request = new Request(Method.POST, url);			
			
			DomRepresentation messageDom = new DomRepresentation(MediaType.ALL);
			messageDom.setDocument(message.getXMLRepresentation());
			request.setEntity(messageDom);
			
			request.setChallengeResponse(challengeResponse);
			Response response = client.handle(request);
			
			Status responseStatus = response.getStatus();
			if(responseStatus.equals(Status.SUCCESS_CREATED))
				return true;
			else 
				throw new APIRequestFailedException(responseStatus);
			
		} catch (IOException e) {
			// creating a DOMRepresentation failed
			return false;
		} catch (ParserConfigurationException e) {
			// creating a DOMRepresentation failed
			return false;
		}
	
	}
	
	/**
	 * Deletes all messages in the queue
	 * 
	 * @return true if all messages were deleted successfully.
	 * @throws APIRequestFailedException If the API request return an unexpected HTTP status code.
	 */
	public boolean clearMessages() throws APIRequestFailedException {
		Client client = new Client(Protocol.HTTP); 
		ChallengeResponse challengeResponse =  new ChallengeResponse(
	    	ChallengeScheme.HTTP_MS_SHAREDKEY,
	    	accountName,
	    	accountSharedKey);
		
		String url = endPoint + "/" + queueName + "/messages";
		Request request = new Request(Method.DELETE, url);
		request.setChallengeResponse(challengeResponse);
		Response response = client.handle(request);
		
		Status responseStatus = response.getStatus();
		if(responseStatus.equals(Status.SUCCESS_NO_CONTENT))
			return true;
		else 
			throw new APIRequestFailedException(responseStatus);
		
	}
	
	public String getQueueName() {
		return this.queueName;
	}

	
	// TODO overload with the visibility timeout argument
	/**
	 * Tries to retrieve the given number of messages.
	 * 
	 * @param nMessages Maximum number of messages to retrieve.
	 * @return The list of messages retrieved.
	 * @throws APIRequestFailedException If the API request return an unexpected HTTP status code.
	 */
	public List<QueueMessage> getMessages(int nMessages) throws APIRequestFailedException {		
				
		try {
			Client client = new Client(Protocol.HTTP); 
			ChallengeResponse challengeResponse =  new ChallengeResponse(
		    	ChallengeScheme.HTTP_MS_SHAREDKEY,
		    	accountName,
		    	accountSharedKey);			
			Map<String, String> urlParamaters = new HashMap<String, String>();
			urlParamaters.put(RequestParams.NumOfMessages, Integer.toString(nMessages));		
			String url = RestHelper.createRequestURI(
						endPoint + "/" + queueName + "/messages", urlParamaters);			
			Request request = new Request(Method.GET, url);
			request.setChallengeResponse(challengeResponse);			
			Response response = client.handle(request);
			// check if the request is successful and extract the message if true.			
			
			if(response.getStatus().isSuccess()){				
				Representation representation = response.getEntity();   			
				//BufferedReader rd = new BufferedReader(new InputStreamReader(representation.getStream(), "UTF-8"));				
				return getMessagesFromResponse(representation.getStream());				
			}
			else
				throw new APIRequestFailedException(response.getStatus());
			
		} catch (UnsupportedEncodingException e) {
			// createRequestURI failed
			return null;
		} catch (IOException e) {
			// getStream() failed
			return null;
		} catch (XPathExpressionException e) {
			//parsing XML failed
			return null;
		} catch (SAXException e) {
			// parsing XML failed
			return null;
		} catch (ParserConfigurationException e) {
			// parsing XML failed
			return null;
		}
		
		
	}
	
	// Parse the queue messages from the XML response.
	private List<QueueMessage> getMessagesFromResponse(InputStream inputStream) 
	throws IOException, SAXException, ParserConfigurationException, XPathExpressionException{	
		
		List<QueueMessage> result = null;
		
		
		BufferedReader rd = new BufferedReader(new InputStreamReader(inputStream, "UTF-8") );			
		rd.skip(1);	        
		Document document=
			DocumentBuilderFactory.newInstance().newDocumentBuilder().
			parse(new InputSource(rd));			
		XPath xp= XPathFactory.newInstance().newXPath();
		NodeList nodelist=(NodeList) xp.evaluate("/"+XMLElements.QueueMessagesList+"/"+XMLElements.QueueMessage,
				document, XPathConstants.NODESET);			
		if(nodelist.getLength()>0){			
			result = new ArrayList<QueueMessage>();				
		}
		for(int i=0;i<nodelist.getLength();i++){		
			QueueMessage qmsg = new QueueMessage();				
			Node messageNode = nodelist.item(i);				
			// retrieve ID
			XPath xpath= XPathFactory.newInstance().newXPath();
			Node idNode=(Node) xp.evaluate(XMLElements.MessageId,
					messageNode, XPathConstants.NODE);				
			qmsg.setMessageId(idNode.getTextContent());				
			// TODO retrieve insertion time, expiration time,pop reciept, etc.				
			Node contentNode=(Node) xp.evaluate(XMLElements.MessageText,
					messageNode, XPathConstants.NODE);
			qmsg.setContent(new String (Base64.decode( contentNode.getTextContent() ) ) );				
			result.add(qmsg);
		}			
				
		return result;
	}
	
	
	
}
